package cn.spdb.harrier.service.db;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import cn.spdb.harrier.common.Constants;
import cn.spdb.harrier.common.enmus.RegistryState;
import cn.spdb.harrier.common.utils.NameThreadFactory;
import cn.spdb.harrier.common.utils.OSUtils;
import cn.spdb.harrier.common.utils.Stopper;
import cn.spdb.harrier.dao.entity.UdsServer;
import cn.spdb.harrier.dao.mapper.UdsServerMapper;
import cn.spdb.harrier.service.registry.HarrierRegistry;
import cn.spdb.harrier.service.registry.SubscribeListener;
import cn.spdb.harrier.service.registry.WatchEvent;

@Component
public class DbRegistryService implements Runnable, HarrierRegistry {

	private Logger logger = LoggerFactory.getLogger(this.getClass());

	private List<SubscribeListener> list = new ArrayList<SubscribeListener>();

	private LinkedBlockingQueue<WatchEvent> eventQueue = new LinkedBlockingQueue<WatchEvent>();

	private ScheduledExecutorService executorTask = Executors.newScheduledThreadPool(1,
			new NameThreadFactory(this.getClass().getSimpleName()));

	private UdsServerMapper serverDB;

	@Value("${harrier.registry.db.heart.seccends:180}")
	private int seccends;
	@Value("${harrier.registry.db.watcher.seccends:10}")
	private int watcherTime = 10;

	private HashSet<UdsServer> udsRegistryServerList = new HashSet<UdsServer>();

	private ExecutorService exec = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(),
			new NameThreadFactory(this.getClass().getSimpleName()));

	private DbRegistryWatch dbRegistryWatch = new DbRegistryWatch();

	@Autowired
	public DbRegistryService(UdsServerMapper serverDB) {
		this.serverDB = serverDB;
		executorTask.scheduleAtFixedRate(dbRegistryWatch, watcherTime, watcherTime, TimeUnit.SECONDS);
	}

	@PostConstruct
	public void start() {
		Thread thread = new Thread(this, this.getClass().getSimpleName());
		thread.setDaemon(true);
		thread.start();
	}

	@PreDestroy
	public void destroy() {
		for (UdsServer udsServer : udsRegistryServerList) {
			udsServer.setLastEnd(LocalDateTime.now());
			udsServer.setIsEnable(false);
			serverDB.updateByIpPortSelective(udsServer);
		}
	}

	public Collection<UdsServer> getUdsServerList() {
		return dbRegistryWatch.getudsServers();
	}

	public UdsServer registrydb(UdsServer udsServer, SubscribeListener listener) {
		udsServer.setIsEnable(true);
		udsServer.setLastStart(LocalDateTime.now());
		registrydb(udsServer);
		addListener(listener);
		udsServer = serverDB.selectByIpPortSelective(udsServer.getIp(), udsServer.getPort()).get();
		udsRegistryServerList.add(udsServer);
		dbRegistryWatch.addUdsServerMap(udsServer);
		WatchEvent watchEvent = new WatchEvent();
		watchEvent.setState(RegistryState.CONNECTED);
		watchEvent.setUdsServer(udsServer);
		eventQueue.add(watchEvent);
		return udsServer;
	}

	private void registrydb(UdsServer udsServer) {
		serverDB.deleteByIpPortSelective(udsServer.getIp(), udsServer.getPort());
		serverDB.insertSelective(udsServer);
	}

	public void unregistrydb(String ip, int port) {
		serverDB.deleteByIpPortSelective(ip, port);
	}

	public void addListener(SubscribeListener listener) {
		if (!ObjectUtils.isEmpty(listener)) {
			list.add(listener);
		}
	}

/**
 * 数据库注册监视器类，负责监控和管理服务节点的注册状态变化
 * 定期检查服务节点的健康状态、连接状态，并生成相应的事件通知
 */
private class DbRegistryWatch implements Runnable {

	/**
	 * 存储已连接服务节点的映射表，key为节点ID，value为UdsServer对象
	 */
	private HashMap<Integer, UdsServer> udsserverMap = new HashMap<Integer, UdsServer>();

	/**
	 * 添加服务节点到本地映射表中
	 *
	 * @param udsServer 需要添加的服务节点对象
	 */
	protected void addUdsServerMap(UdsServer udsServer) {
		udsserverMap.put(udsServer.getId(), udsServer);
	}

	/**
	 * 获取所有已连接的服务节点集合
	 *
	 * @return 服务节点集合
	 */
	protected Collection<UdsServer> getudsServers() {
		return udsserverMap.values();
	}

	/**
	 * 执行监控任务，定期检查服务节点状态并生成事件
	 * 主要功能包括：
	 * 1. 更新本地注册服务的心跳时间
	 * 2. 收集服务节点的CPU和内存使用情况
	 * 3. 检查服务节点的连接状态变化（连接、重连、断开、暂停等）
	 * 4. 生成相应的WatchEvent事件并加入事件队列
	 */
	@Override
	public void run() {
		try {
			if (ObjectUtils.isEmpty(udsRegistryServerList)) {
				return;
			}
			// 所有服务集合
			List<UdsServer> servers = new ArrayList<UdsServer>();
			WatchEvent watchEvent = null;
			for (UdsServer udsServer : udsRegistryServerList) {
				// 更新本地注册服务的时间
				udsServer.setUpdateTime(LocalDateTime.now());
				try {
					udsServer.addPara(Constants.CPU,
							new Object[] { Runtime.getRuntime().availableProcessors(), OSUtils.cpuUsage() });
					udsServer.addPara(Constants.MEM,
							new Object[] { OSUtils.totalPhysicalMemorySize(), OSUtils.memoryUsage() });
					// 服务心跳
					serverDB.updateByIpPortSelective(udsServer);
				} catch (Exception e) {
					watchEvent = new WatchEvent();
					watchEvent.setState(RegistryState.SUSPENDED);
					watchEvent.setUdsServer(udsServer);
					logger.info("local suspended node {}", udsServer.getServerName());
					e.printStackTrace();
				}
				// 获取所有相关集合的服务
				List<UdsServer> serverList = serverDB.selectListByRole(udsServer.getServerRoleName(),
						udsServer.getServerRoleNameGroup(), true);
				servers.addAll(serverList);
				Iterator<UdsServer> iterator = serverList.iterator();
				// 判断各个服务的状态
				while (iterator.hasNext()) {
					UdsServer us = iterator.next();
					// 本地参数是否更新
					if (us.equals(udsServer)) {
						if (!us.getPara().equals(udsServer.getPara())) {
							watchEvent = new WatchEvent();
							watchEvent.setState(RegistryState.UPDATE_PARA);
							watchEvent.setUdsServer(us);
							udsserverMap.put(us.getId(), us);
						}
						continue;
					}

					// 从本地映射表中获取对应ID的服务节点
					UdsServer localUs = udsserverMap.get(us.getId());
					// 计算该服务节点距离现在的时间差，判断是否超时
					Duration duration = Duration.between(us.getUpdateTime(), LocalDateTime.now());
					// 判断服务节点是否在有效时间内且处于启用状态
					Boolean usUse = duration.getSeconds() < seccends && us.getIsEnable();

					// 如果本地映射表中没有该服务节点的记录
					if (localUs == null) {
					    // 如果该服务节点是可用的（未超时且启用），则认为是新连接的节点
					    if (usUse) {
					        watchEvent = new WatchEvent();
					        watchEvent.setState(RegistryState.CONNECTED);  // 设置事件状态为连接
					        watchEvent.setUdsServer(us);  // 设置事件关联的服务节点
					        udsserverMap.put(us.getId(), us);  // 将该节点加入本地映射表
					        logger.info("local {} node connected {}", udsServer.getServerName(),
					                us.getServerName());  // 记录连接日志
					    }
					} else {
					    // 如果本地映射表中已有该服务节点记录
					    // 计算本地记录的服务节点的时间差
					    Duration localDuration = Duration.between(localUs.getUpdateTime(), LocalDateTime.now());
					    // 判断本地记录的服务节点是否在有效时间内且处于启用状态
					    Boolean localUse = localDuration.getSeconds() < seccends && localUs.getIsEnable();

					    // 如果数据库中的服务节点是可用的
					    if (usUse) {
					        // 如果本地记录的服务节点也是可用的
					        if (localUse) {
					            // 检查参数是否发生变化
					            if (!us.getPara().equals(localUs.getPara())) {
					                watchEvent = new WatchEvent();
					                watchEvent.setState(RegistryState.UPDATE_PARA);  // 设置事件状态为参数更新
					                watchEvent.setUdsServer(us);  // 设置事件关联的服务节点
					            }
					            udsserverMap.put(us.getId(), us);  // 更新本地映射表中的记录
					        } else {
					            // 如果本地记录的服务节点之前不可用，现在变为可用，则认为是重连
					            watchEvent = new WatchEvent();
					            watchEvent.setState(RegistryState.RECONNECTED);  // 设置事件状态为重连
					            watchEvent.setUdsServer(us);  // 设置事件关联的服务节点
					            udsserverMap.put(us.getId(), us);  // 更新本地映射表中的记录
					            logger.info("local {} node reconnected {}", udsServer.getServerName(),
					                    us.getServerName());  // 记录重连日志
					        }
					    } else {
					        // 如果数据库中的服务节点不可用
					        // 且本地记录的服务节点是启用状态，则认为服务节点已暂停
					        if (localUs.getIsEnable()) {
					            watchEvent = new WatchEvent();
					            watchEvent.setState(RegistryState.SUSPENDED);  // 设置事件状态为暂停
					            watchEvent.setUdsServer(us);  // 设置事件关联的服务节点
					            localUs.setIsEnable(false);  // 设置本地记录为禁用状态
					            logger.info("local {} node suspended {}", udsServer.getServerName(),
					                    us.getServerName());  // 记录暂停日志
					        }
					    }
					}
					if (watchEvent != null) {
						eventQueue.add(watchEvent);
					}
				}
			}

			Iterator<UdsServer> it = udsserverMap.values().iterator();
			while (it.hasNext()) {
				UdsServer localus = it.next();
				if (!servers.contains(localus)) {
					// 移除
					it.remove();
					watchEvent = new WatchEvent();
					watchEvent.setState(RegistryState.DISCONNECTED);
					watchEvent.setUdsServer(localus);
					eventQueue.add(watchEvent);
					logger.info("node disconnected {}", localus);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}


/**
 * 持续从事件队列中取出 WatchEvent 事件，并通知所有订阅该事件的监听器。
 * 使用线程池异步执行每个监听器的回调，确保非阻塞处理事件。
 */
@Override
public void run() {
    while (Stopper.isRunning()) {
        try {
            WatchEvent watchEvent = eventQueue.take(); // 从队列中取出事件
            if (watchEvent != null) {
                for (SubscribeListener listener : list) { // 遍历所有订阅者
                    try {
                        // 异步执行监听器的回调函数，避免阻塞当前线程
                        exec.execute(() -> listener.notity(watchEvent));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}


	public int getWatcherTime() {
		return watcherTime;
	}

	public void setWatcherTime(int watcherTime) {
		this.watcherTime = watcherTime;
	}

/**
 * 根据指定的节点集群类型获取当前可用的 Leader 节点。
 * Leader 是根据 ID 最小原则选取的处于启用状态的服务节点。
 *
 * @param nodeClusterType 节点集群类型（如 master、alarm 等）
 * @return 当前符合条件的 Leader 节点，如果没有可用节点则返回 null
 */
public UdsServer getLeader(String nodeClusterType) {
    UdsServer tmp = null;
    for (UdsServer udsServer : getUdsServerList()) {
        // 匹配相同集群类型且启用状态的服务节点
        if (udsServer.getNodeClusterType().equals(nodeClusterType)) {
            if (!udsServer.getIsEnable()) {
                continue; // 跳过未启用的节点
            }
            // 选择 ID 更小的节点作为 Leader
            if (ObjectUtils.isEmpty(tmp) || tmp.getId() > udsServer.getId()) {
                tmp = udsServer;
            }
        }
    }
    return tmp;
}


	public UdsServer getMaster() {
		return getLeader(Constants.THREAD_NAME_MASTER_SERVER);
	}

	public UdsServer getAlarm() {
		return getLeader(Constants.THREAD_NAME_ALARM_SERVER);
	}

}
